package com.liyueheng.app.operationalDataStore

import com.liyueheng.util.{ConfigLoader, SaveAsTable, SparkConf}
import org.apache.spark.sql.functions.col

object DataCleaner {

  // 清洗 gmall_ods库中的用户数据表(user)，并将清洗完毕的数据写入到 gmall_dwd库中的用户数据表(user)
  def cleanerOdsUserTable(): Unit = {

    // 读取配置文件中的表名配置
    val tableName = ConfigLoader.getString("tables.ods.clean").split(",").toList // 获取表名并转换为 List
    val databaseOds = ConfigLoader.getString("databases.ods") // 获取 ods 数据库名
    val databaseDwd = ConfigLoader.getString("databases.dwd") // 获取 dwd 数据库名
    val appName = ConfigLoader.getString("app.name") // 获取应用名称
    for (elem <- tableName) {
      val spark = SparkConf.createSparkSession(appName)

      try {
        println(s"--------- 处理$databaseOds." + elem + " 表中重复、空的字段 ----------")
        // 获取数据库中的表
        val tableDF = spark.table(s"$databaseOds.$elem")
        // 统计去重前的总行数
        val tableDFCount = tableDF.count()
        /*
         * 去重复行
         */
        val tableCleanDF = tableDF.dropDuplicates()
        val tableCleanCount = tableCleanDF.count()
        println(s"原始行数:$tableDFCount, 去重后行数:$tableCleanCount")
        /*
         * 去除空行
         */
        // 获取表的所有列名
        val columns = tableDF.columns

        // 动态生成去空值的条件
        val nonNullConditions = columns.map(colName => col(colName).isNotNull).reduce(_ && _)

        // 使用动态生成的条件去除空值行
        val tableDropNull = tableCleanDF.filter(nonNullConditions)
        val afterDropNullCount = tableDropNull.count()
        println(s"原始行数:$tableCleanCount, 去空后行数:$afterDropNullCount")
        // 打印前10行数据
        tableDropNull.show(10)
        // 保存到目标库中的表
        tableDropNull.write.mode("overwrite").format("hive").saveAsTable(s"$databaseDwd.$elem")
        SaveAsTable.saveAsTable(tableDropNull,s"$databaseDwd.$elem")
        // 释放缓存
        tableDF.unpersist()
        tableDropNull.unpersist()
      }finally{
        SparkConf.stopSparkSession(spark)
      }
    }
  }

}
